#include "timer_log.h"
#include "nccl.h"
#include "core.h"
#include <sys/un.h>
#include <iostream>
#include <fstream>
#include <string>
#include <ctime>
#include <iomanip>
#include <sstream>


// void sigpipe_handler(int signum) {
//   ;
// }
std::string getCurrentTimeString() {
    std::time_t now = std::time(nullptr);
    std::tm* localTime = std::localtime(&now);

    std::ostringstream oss;
    oss << std::put_time(localTime, "%Y%m%d_%H%M%S");
    return oss.str();
}

void printLogInfo(struct timer_log log){
  INFO(NCCL_NET, "%d.%d.%d.%d->%d.%d.%d.%d send %d Bits used %lld nsec", 
             log.srcIp[0],log.srcIp[1],log.srcIp[2],log.srcIp[3],
             log.dscIp[0],log.dscIp[1],log.dscIp[2],log.dscIp[3],
             log.size,log.diff
  );
}


void* timerLogService(void *args){
  // signal(SIGPIPE, sigpipe_handler);

  int client_fd;
  struct sockaddr_un server_addr;
  memset(&server_addr, 0, sizeof(server_addr));
  server_addr.sun_family = AF_UNIX;
  strncpy(server_addr.sun_path, SOCK_PATH, sizeof(server_addr.sun_path) - 1);

  if(TIMER_LOG_NCCL_TELEMETRY){
    char buffer[256];
std::ofstream logFile;
  //std::string baseName = global_timer_log.log[i].NetworkCardName;
  std::string timestamp = getCurrentTimeString();

    while(!global_timer_log.stop){
      global_timer_log.collect = 1;
      if(global_timer_log.cnt == TIMER_LOG_MAX_LEN){
        while(global_timer_log.setState(TIMER_LOG_QUEUE_READ) == 0);
        if(global_timer_log.cnt == TIMER_LOG_MAX_LEN){
          global_timer_log.collect = 0;
          int cnt = TIMER_LOG_MAX_LEN;
          // if(send(client_fd, &cnt, sizeof(int), 0) == -1)  INFO(NCCL_INIT, "TIMER_LOG  SEND ERROR");
          __sync_synchronize();
          for(int i = 0; i < cnt; i++) {
            memcpy(global_timer_log.t_res[i].srcIp, global_timer_log.log->srcIp, 4*sizeof(u_int8_t));
            memcpy(global_timer_log.t_res[i].dscIp, global_timer_log.log->dscIp, 4*sizeof(u_int8_t));
            global_timer_log.t_res[i].send_start_sec = 1000000000L * global_timer_log.log[i].send_start.tv_sec + global_timer_log.log[i].send_start.tv_nsec;
            global_timer_log.t_res[i].send_end_sec   = 1000000000L * global_timer_log.log[i].send_end.tv_sec + global_timer_log.log[i].send_end.tv_nsec;
            global_timer_log.t_res[i].diff           = global_timer_log.log[i].diff;
            global_timer_log.t_res[i].size           = global_timer_log.log[i].size;
            sprintf(buffer, "%d.%d.%d.%d->%d.%d.%d.%d send %d Bytes used %lld nsec", 
            global_timer_log.t_res[i].srcIp[0],global_timer_log.t_res[i].srcIp[1],global_timer_log.t_res[i].srcIp[2],global_timer_log.t_res[i].srcIp[3],
            global_timer_log.t_res[i].dscIp[0],global_timer_log.t_res[i].dscIp[1],global_timer_log.t_res[i].dscIp[2],global_timer_log.t_res[i].dscIp[3],
            global_timer_log.t_res[i].size,global_timer_log.t_res[i].diff
            );
            INFO(NCCL_NET, "%s", buffer);
          
          //INFO(NCCL_INIT,"---------RUNNING AT LINE: %d ,FUNCTION: %s",__LINE__,__func__);
          //INFO(NCCL_INIT,"---------std::ofstream logFile;____________");
          // 清理 buffer 中的空字符
          std::string cleanedBuffer = buffer;
          cleanedBuffer.erase(std::remove(cleanedBuffer.begin(), cleanedBuffer.end(), '\0'), cleanedBuffer.end());
          char hostname[1024];
          getHostName(hostname, 1024, '.');
          std::string str(hostname);
          std::string basename =str +"_" + global_timer_log.log[i].NetworkCardName;
          std::string logDirectory = "./LOG/";
    //std::string logDirectory = "";
    
          std::string logFileName=logDirectory + basename+"_" + timestamp + ".log";
          logFile.open(logFileName, std::ios_base::app);
          if (logFile.is_open()) {
            logFile << str +":" + global_timer_log.log[i].NetworkCardName<<":"<<cleanedBuffer<< std::endl;
            logFile.close();
          } else {
            WARN("Unable to open log file %s.",logFileName.c_str()) ;
          }
          INFO(NCCL_INIT,"logFileName:%s",logFileName.c_str());
        }
        global_timer_log.cnt = 0;
        }
        global_timer_log.freeState();
      }
      sleep(3);
    }
    
    while(!global_timer_log.stop){
      if ((client_fd = socket(AF_UNIX, SOCK_STREAM, 0)) == -1){
        sleep(3);
        continue;
      }
      struct timeval timeout;
      timeout.tv_sec = 3;
      timeout.tv_usec = 0;
      if (setsockopt(client_fd, SOL_SOCKET, SO_RCVTIMEO, (char *)&timeout, sizeof(timeout)) < 0) {
        close(client_fd);
        sleep(3);
        continue;
      }

      while (!global_timer_log.stop && connect(client_fd, (struct sockaddr *)&server_addr, sizeof(server_addr)) == -1) {
        // INFO(NCCL_INIT, "sock domain connecting");
        sleep(3);
      }

      int recv_flag = 0;
      recv(client_fd, &recv_flag, sizeof(recv_flag), MSG_NOSIGNAL | MSG_WAITALL);
      // INFO(NCCL_INIT, "recv_flag = %d", recv_flag);
      // recv_flag = 3;
      if(recv_flag){
        global_timer_log.collect = 1;
      }
      else{
        global_timer_log.collect = 0;
        close(client_fd);
        sleep(3);
        continue;
      }

      
      while(!global_timer_log.stop){
        if(global_timer_log.cnt == TIMER_LOG_MAX_LEN){
          while(global_timer_log.setState(TIMER_LOG_QUEUE_READ) == 0);

          if(global_timer_log.cnt == TIMER_LOG_MAX_LEN){
            global_timer_log.collect = 0;
            int cnt = TIMER_LOG_MAX_LEN;
            // if(send(client_fd, &cnt, sizeof(int), 0) == -1)  INFO(NCCL_INIT, "TIMER_LOG  SEND ERROR");
            __sync_synchronize();
            for(int i = 0; i < cnt; i++){
              memcpy(global_timer_log.t_res[i].srcIp, global_timer_log.log->srcIp, 4*sizeof(u_int8_t));
              memcpy(global_timer_log.t_res[i].dscIp, global_timer_log.log->dscIp, 4*sizeof(u_int8_t));
              global_timer_log.t_res[i].send_start_sec = 1000000000L * global_timer_log.log[i].send_start.tv_sec + global_timer_log.log[i].send_start.tv_nsec;
              global_timer_log.t_res[i].send_end_sec   = 1000000000L * global_timer_log.log[i].send_end.tv_sec + global_timer_log.log[i].send_end.tv_nsec;
              global_timer_log.t_res[i].diff           = global_timer_log.log[i].diff;
              global_timer_log.t_res[i].size           = global_timer_log.log[i].size;
              // INFO(NCCL_INIT, "%d.%d.%d.%d", global_timer_log.t_res[i].srcIp[0],global_timer_log.t_res[i].srcIp[1],global_timer_log.t_res[i].srcIp[2],global_timer_log.t_res[i].srcIp[3]);
            }
            send(client_fd, global_timer_log.t_res, sizeof(struct timer_log_res) * TIMER_LOG_MAX_LEN, MSG_NOSIGNAL);
            global_timer_log.cnt = 0;
            global_timer_log.freeState();
            break;
          }
          
          global_timer_log.freeState();
        }
      }
      //配置发送频率
      close(client_fd);
      sleep(recv_flag);
    }
  }
  return 0;
}
